#ifndef __MQ_CHANNEL_HPP__
#define __MQ_CHANNEL_HPP__

#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_log.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "mq_router.hpp"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "mq_consume.hpp"
#include "mq_virtualhost.hpp"

namespace mq
{
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    //using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;

    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    class Channel
    {
    public:
        using Ptr = std::shared_ptr<Channel>;

        Channel(const std::string& id,
                const VirtualHost::Ptr& host,
                const ConsumerManager::Ptr cmp,
                const ProtobufCodecPtr& codec,
                const muduo::net::TcpConnectionPtr& conn,
                const threadpool::Ptr& pool)
            :_cid(id),
            _host(host),
            _cmp(cmp),
            _codec(codec),
            _conn(conn),
            _pool(pool)
        {
            DLOG("new Channel:%p",this);
        }
        ~Channel()
        {
            if(_consumer.get() != nullptr)
            {
                // 删除订阅对象。
                _cmp->remove(_consumer->tag,_consumer->qname);
                
            }
        }
        // 交换机声明与删除。
        void declareExchange(const declareExchangeRequestPtr& req)
        {
            bool ret = _host->declareExchange(req->exchange_name(),
                        req->exchange_type(),req->durable(),
                        req->auto_delete(),req->args()
                        );

            return basicResponse(ret,req->rid(),req->cid());
        }
        void deleteExchange(const deleteExchangeRequestPtr& req)
        {
            _host->deleteExchange(req->exchange_name());

            return basicResponse(true,req->rid(),req->cid());
        }
        // 队列的声明和删除。
        void declareMsgqueue(const declareQueueRequestPtr& req)
        {
            bool ret = _host->declareMsgqueue(req->queue_name(),req->durable(),req->exculsive(),
                                              req->auto_delete(),req->args());
            
            if(ret == false)
            {
                // DLOG("Channel:Declare Queue Failed");

                return basicResponse(false,req->rid(),req->cid());
            }

            DLOG("Declare Queue Successful");
            
            _cmp->initQueueConsumer(req->queue_name());

            return basicResponse(true,req->rid(),req->cid());
        }
        void deleteMsgqueue(const deleteQueueRequestPtr& req)
        {
            _host->deleteQueue(req->queue_name());
            _cmp->destroyQueueConsumer(req->queue_name());

            return basicResponse(true,req->rid(),req->cid());
        }
        // 绑定和解除绑定。
        void queueBind(const queueBindRequestPtr& req)
        {
            bool ret = _host->bind(req->exchange_name(),req->queue_name(),req->bind_key());

            return basicResponse(ret,req->rid(),req->cid());
        }
        void queueUnbind(const queueUnBindRequestPtr& req)
        {
            _host->unbind(req->exchange_name(),req->queue_name());

            return basicResponse(true,req->rid(),req->cid());
        }
        // 消息的发布。
        void basicPublish(const basicPublishRequestPtr& req)
        {
            // DLOG("消息开始发布。");
            // 1.判断交换机是否存在。
            auto ep = _host->selectExchange(req->exchange_name());
            if(ep.get() == nullptr)
            {
                DLOG("basicPublish消息发布失败");

                return basicResponse(false,req->rid(),req->cid());
            }
            // 2.进行交换路由，判断消息可以发布到交换机绑定的那个队列中。      exchange_name + bind
            MsgQueueBindingMap mqbm = _host->exchangeBindings(req->exchange_name());
            BasicProperties* properties = nullptr;
            std::string router_key;
            if(req->has_properties())
            {
                properties = req->mutable_properties();
                router_key = req->properties().rounting_key();
            }
            DLOG("消息发布MsgQueueBindingMap.size() = %ld",mqbm.size());
            for(auto& it : mqbm)
            {
                DLOG("[bind queue_name:%s]",it.first.c_str());
            }
            for(auto& binding : mqbm)
            {
                DLOG("etype:%d \t router_key:%s \t bind_key:%s",
                    ep->type,
                    router_key.c_str(),
                    binding.second->bind_key.c_str());
                bool ret = Router::route(ep->type,binding.second->bind_key,router_key);
                DLOG("route:%d",ret);
                // if(Router::route(ep->type,router_key,binding.second->bind_key) == true)
                if(ret)
                {
                    DLOG("_host->basicPublish");
                    // 3.将消息添加到队列中(添加消息的管理)
                    _host->basicPublish(binding.first,properties,req->body());
                    // 4.向线程池中添加一个消息消费任务(向指定队列的订阅者去推送消息)
                    auto task = std::bind(&Channel::consume,this,binding.first);
                    _pool->push(task);
                }
            }
            // DLOG("basicPublish消息发布成功");

            return basicResponse(true,req->rid(),req->cid());
        }
        // 消息的确认。
        void basicAck(const basicAckRequestPtr& req)
        {
            _host->basicAck(req->queue_name(),req->message_id());

            return basicResponse(true,req->rid(),req->cid());
        }
        // 订阅消息队列。
        void basicConsume(const basicConsumeRequestPtr& req)
        {
            // 1.判断队列是否存在。
            DLOG("_host->_mmp.total_count():%ld",_host->_mmp->total_count("queue1"));
            DLOG("_host->_mmp.getable_count():%ld",_host->_mmp->getable_count("queue1"));
            bool ret = _host->existMsgQueue(req->queue_name());

            if(ret == false)
            {
                DLOG("队列不存在%s",req->queue_name().c_str());

                return basicResponse(false,req->rid(),req->cid());
            }

            // DLOG("host队列%s存在。",req->queue_name().c_str());

            // 2.创建队列的消费者。
            auto cb = std::bind(&Channel::callback,this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
            // 创建好消费者后。当前Channel就是一个消费者。
            DLOG("Build _consumer!");
            DLOG("consumer_tag:%s \t queue_name:%s \t auto_ack:%d",req->consumer_tag().c_str(),req->queue_name().c_str(),req->auto_ack());
            _consumer = _cmp->create(req->consumer_tag(),req->queue_name(),req->auto_ack(),std::move(cb));
            DLOG("_consumer:%p",_consumer.get());

            return basicResponse(true,req->rid(),req->cid());
        }
        // 取消订阅消息队列。
        void basicCancel(const basicCancelRequestPtr& req)
        {
            DLOG("取消订阅的时候找不到消费者信息。");
            _cmp->remove(req->consumer_tag(),req->queue_name());

            return basicResponse(true,req->rid(),req->cid());
        }
    private:
        void callback(const std::string& tag,const BasicProperties* bp,const std::string& body)
        {
            basicConsumeResponse resp;

            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);

            if(bp != nullptr)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_rounting_key(bp->rounting_key());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            }

            _codec->send(_conn,resp);
        }
        void consume(const std::string &qname)
        {
            // 指定队列消费信息。
            // 1.从队列中取出一条消息。（同名）
            DLOG("从队列中取出一条消息。");
            MessagePtr mp = _host->basicConsume(qname);
            DLOG("MessagePtr:%p",mp.get());
            if(mp.get() == nullptr)
            {
                DLOG("消息队列%s中没有消息。",qname.c_str());

                return ;
            }
            // 2.从订阅者队列中取出一个订阅者。（同名）
            Consumer::Ptr cp = _cmp->choose(qname);
            if(cp.get() == nullptr)
            {
                DLOG("consume failed,%s can't find !",qname.c_str());

                return ;
            }
            DLOG("ConsumePtr:%p",cp.get());
            DLOG("qname:%s \n tag:%s \n ",cp->qname.c_str(),cp->tag.c_str());
            // 3.调用订阅者对应的消息处理函数，实现消息的推送。
            cp->callback(cp->tag,mp->mutable_payload()->mutable_properties(),mp->payload().body());
            // 4.判断如果订阅者是自动确认---不需要等待确认，直接删除消息。
            // 否则需要外部收到消息确认后再删除。
            if(cp->auto_ack)
            {
                _host->basicAck(qname,mp->payload().properties().id());

                return ;
            }

            return ;
        }
        void basicResponse(bool ok,const std::string& rid,const std::string& cid)
        {
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn,resp);
        }
    private:
        // 序列号。
        std::string _cid;
        // 虚拟机。
        VirtualHost::Ptr _host;
        // 消费者管理。
        ConsumerManager::Ptr _cmp;
        // protobuf协议管理。
        ProtobufCodecPtr _codec;
        // 信道关联的消费者。
        Consumer::Ptr _consumer;
        // Tcp网络协议。
        muduo::net::TcpConnectionPtr _conn;
        // 线程池。
        threadpool::Ptr _pool;
    };
    class ChannelManager
    {
    public:
            using Ptr = std::shared_ptr<ChannelManager>;
            ChannelManager(){ DLOG("channel_manager:%p",this) }
            ~ChannelManager(){}
            bool openChannel(const std::string& id,
                const VirtualHost::Ptr& host,
                const ConsumerManager::Ptr cmp,
                const ProtobufCodecPtr& codec,
                const muduo::net::TcpConnectionPtr& conn,
                const threadpool::Ptr& pool)
            {
                std::unique_lock<std::mutex> lock(_mutex);

                auto it = _channels.find(id);
                // 已经存在则不需要再创建了。
                if(it != _channels.end())
                {
                    return false;
                }
                auto channel = std::make_shared<Channel>(id,host,cmp,codec,conn,pool);
                _channels.insert(std::make_pair(id,channel));
                return true;
            }
            void closeChannel(const std::string& id)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                
                _channels.erase(id);
            }
            Channel::Ptr choose(const std::string& id)
            {
                auto channel = _channels.find(id);

                // 不存在该信道。
                if(channel == _channels.end())
                {
                    DLOG("不存在id:%s的信道",id.c_str());

                    return Channel::Ptr();
                }

                return channel->second;
            }
        private:
            std::mutex _mutex;
            std::unordered_map<std::string,Channel::Ptr> _channels;
    };
}

#endif
